package com.hzh.flink.sink
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo4MysqlSink {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)
    //统计单词的数量
    val countDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)

    //将统计好的结果保存到mysql中
    countDS.addSink(new MysqlSink)

    env.execute()

  }

  /**
   * 自定斯诺克将数据保存到mysql中
   * SinkFunction:
   * RichSinkFunction: 多个open和close方法
   *
   */

  class MysqlSink extends RichSinkFunction[(String, Int)] {

    var con: Connection = _
    var stat: PreparedStatement = _
    /**
     * open： 在invoke之前执行，每一个task中只执行一次
     * 一般用于初始化数据库链接
     *
     */
    override def open(parameters: Configuration): Unit = {
      //1、加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      //创建链接
      con = DriverManager.getConnection("jdbc:mysql://master:3306/test", "root", "123456")
      //编写插入数据的sql
      //replace :如果不存在插入，如果存在就替换，需要在表中设置主键
      stat = con.prepareStatement("replace into word_count(words,count) values(?,?)")

    }
    /**
     * 任务关闭的时候执行，一般用于回收资源
     *
     */
    override def close(): Unit = {
      //关闭链接
      stat.close()
      con.close()
    }
    /**
     * 每一条数据会执行一次
     * 使用jdbc将数据保存到mysql中
     *
     * @param kv      ： 一行数据
     * @param context ： 上下文对象
     */
    override def invoke(kv: (String, Int), context: SinkFunction.Context): Unit = {
      //设置参数
      stat.setString(1, kv._1)
      stat.setInt(2, kv._2)
      //执行插入
      stat.execute()
    }
  }
}